package wnco.rich;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import wnco.base.SomData;

public class RichFunctionW {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<SomData> dss = env.fromElements(
                new SomData("aa", 1, 496656L),
                new SomData("bb", 2, 745451L),
                new SomData("cc", 3, 351548L),
                new SomData("dd", 4, 126988L),
                new SomData("ee", 5, 984656L));

        SingleOutputStreamOperator<Integer> map = dss.map(new RichMapFunction<SomData, Integer>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("open");
            }

            @Override
            public Integer map(SomData value) throws Exception {
                return value.getAge();
            }

            @Override
            public void close() throws Exception {
                super.close();

                System.out.println("close".hashCode());
            }
        });
        map.print();
        env.execute();
    }
}

